<?php
#THIS JOB INSERTS DATA DIRECTLY INTO the tmp_shard
#This job also uses an unbuffered query
require_once 'worker_common.php';

if(function_exists('gc_enable')) gc_enable();


class Net_Gearman_Job_store_resultset extends Net_Gearman_Job_Common
{
    public function run($arg) {

	$errors = false;
	$stmt = false;
	if(!$arg) return;

	$arg = (object)$arg;

	#broadcast simple queries like DML and DDL
	if($arg->action == 'no_resultset') {
		return $this->broadcast($arg);
	}

	#store the resultset in a table
	$conn = SimpleDAL::factory($arg->shard);
	if($conn->my_error()) { 
		return(return_error('Failed to connect to storage node',$arg->shard,$conn->my_error()));
	}

	$tmp_conn = SimpleDAL::factory($arg->tmp_shard);
	if(!$tmp_conn) return(return_error('Failed to connect to coordination node', $arg->tmp_shard, $tmp_conn->my_error(),$arg->query_id));

	if(!$conn->my_select_db($arg->shard['db'])) return(return_error('Could not change to database on storage node', $arg->shard, $conn->my_error(),$arg->query_id));
	if(!$tmp_conn->my_select_db($arg->tmp_shard['db'])) return(return_error('Could not change to database on coordinator node', $arg->tmp_shard, $tmp_conn->my_error(),$arg->query_id));

	$stmt = $tmp_conn->my_query("select @@max_allowed_packet");
	if(!$stmt) return(return_error('Could not get max_allowed_packet', $this->tmp_shard, $tmp_conn->my_error(),$arg->query_id));
	$row = $tmp_conn->my_fetch_array();
	$tmp_conn->my_free_result();
	$max_len = $row[0] - 4096;

	#FIXME: use the engine specified in the config
	$stmt = $tmp_conn->my_query("SET STORAGE_ENGINE=MYISAM");

	$sql = "select count(*) cnt from information_schema.tables where table_schema = '" . $tmp_conn->my_escape_string($arg->tmp_shard['db']) . "' and table_name = '" . $arg->table_name . "'";
	$tmp_conn->my_query($sql);
	$row = $tmp_conn->my_fetch_assoc();
	
	$table_exists = ($row !== false && $row['cnt'] > 0);
	$tmp_conn->my_free_result();

	#get the data from the shard using MYSQL_USE_RESULT

	#echo "RUNNING:{$arg->sql}\n";
	$stmt = $conn->my_unbuffered_query($arg->sql);    
	if($conn->my_error()) {
		return(return_error('Failed to execute query on node',$arg->shard,$conn->my_error(),$arg->query_id));
	}

	$created_sql = false;	
	$sql = "INSERT INTO `{$arg->table_name}` VALUES ";
	$values = "";
	$odku="";
	if(!empty($arg->coord_odku))  {
		$odku=' ON DUPLICATE KEY UPDATE ' . join(",\n",$arg->coord_odku);
		
	}
	$got_rows = 0;
	$hashes['gb_hash'] = array();

	while($stmt && $row = $conn->my_fetch_assoc($stmt)) {
		if(!empty($row['gb_hash'])) $hashes['gb_hash'][$row['gb_hash']] = 1;
		++$got_rows;

		#this has to be in the loop becuse we get the structure from the first row
		if(!$created_sql)  {
                	
                	$created_sql = "CREATE TABLE IF NOT EXISTS {$arg->table_name} (";
	                $col_sql = "";
        	        $cols = array_keys($row);

	                for($i = 0; $i<count($cols);++$i) {
				$use_datatype="VARBINARY(255)";
				switch($conn->my_field_type($stmt,$i)) {
					case 'int':
					$use_datatype="BIGINT";
					break;

					case 'real':
					$use_datatype="DECIMAL(20,6)";
					break;
	
					case 'string':
					$use_datatype="VARBINARY(255)";
					break;

					case 'blob':
					case 'default':
					$use_datatype='VARBINARY(255)';
				}

        	                if($cols[$i] == "0") continue;
	                        if($col_sql) $col_sql .= ", ";
	                        $col_sql .= "`{$cols[$i]}` $use_datatype";

	                }
	                $created_sql .= $col_sql; 

                	if(stristr($arg->sql, 'GROUP BY')) { 
				if(!empty($arg->agg_key_cols) && $arg->agg_key_cols) $created_sql .= ",PRIMARY KEY(" . $arg->agg_key_cols . ") ";
			}

			$created_sql .= ") ENGINE=". $arg->engine;

                	if(!$table_exists) {
				#echo $created_sql ."\n";
				if(!$tmp_conn->my_query($created_sql)) {
					return(return_error('Could not create coordination table {' . $created_sql . "}", $this->tmp_shard, $tmp_conn->my_error(),$arg->query_id));
				}
			}
		}

		if($values) $values .= ",";
		$val_list = "";
		foreach($row as $col => $val) {
			#echo "COL: $col, VAL: $val\n";
			if($col == "0") continue;
			if($val_list) $val_list .= ',';
			if(is_numeric($val)) {
				 $val_list .= "{$val}";
			} else {
				$val_list .= "'" . $tmp_conn->my_real_escape_string($val) . "'";
			}
		}
		$values .= "({$val_list})";

		if(strlen($values) >= $max_len) {
			#FIXME
			#we should only issue this for MyISAM tables
			$tmp_conn->my_query("LOCK TABLES `{$arg->table_name}` WRITE");
			if(!$tmp_conn->my_query($sql . $values . $odku)) {
				echo "FAILED\n";
				return(return_error('Error while inserting' . $sql . $values . $odku, $arg->tmp_shard, $tmp_conn->my_error(),$arg->query_id));
			}
			$values = "";

		}
	}

	#any rows left over?
	if($values) {
		if(!$tmp_conn->my_query($sql . $values . $odku)) {
			return(return_error('Error while inserting' . $sql . $values . $odku, $arg->tmp_shard, $tmp_conn->my_error(),$arg->query_id));
		}
	}

	if($stmt) $conn->my_free_result($stmt);

	#FIXME: only MyISAM
	$sql = "UNLOCK TABLES";
	$tmp_conn->my_query($sql);

	#free resources
	unset($tmp_conn);
	unset($conn);
	unset($sql);
	unset($values);
	if(function_exists('gc_collect_cycles')) gc_collect_cycles();
	
    	return(return_ok($got_rows, $hashes,$arg->query_id));
    }

	private function broadcast($arg) {
		if(!$arg) return;
        	$arg = (object)$arg;

        	$conn = SimpleDAL::factory($arg->shard);
        	if($conn->my_error()) {
        	        return(return_message($conn->my_error(),$arg->shard,$arg->query_id));
        	}

        	if(!$conn->my_select_db($arg->shard['db'])) {
        	        return(return_message($conn->my_error(),$arg->shard,$arg->query_id));
        	}

        	$stmt = $conn->my_query($arg->sql);
        	if(!$stmt) {
        	        return(return_message($conn->my_error(),$arg->shard,$arg->query_id));
        	}

        	$last_insert_id = 0;
        	if($affected_row_count = $conn->my_affected_rows() == 0) $affected_row_count="0";


        	if(preg_match('/insert |last_insert_id/', $arg->sql)) {
        	        $last_insert_id = $conn->my_insert_id();
        	}

        	unset($conn);

        	return(return_message('success', $arg->shard, 'info', $affected_row_count, $last_insert_id,$arg->query_id));

	}
}

?>
